package cn.caplike.demo.project.reactor.quickstart;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.atomic.AtomicLong;

/**
 * Description: 以编程方式创建一个序列<br>
 * Reference: <a href="https://projectreactor.io/docs/core/release/reference/#producing">Programmatically creating a sequence</a>
 *
 * @author LiKe
 * @version 1.0.0
 * @date 2020-08-20 15:47
 */
public class ProgrammaticallyCreatingSequence {

    // ~ 4.4.1 Synchronous generate
    //   https://projectreactor.io/docs/core/release/reference/#producing.generate

    //   同步的一个一个的生成, 意味着 sink 本质上是一个 reactor.core.publisher.SynchronousSink<String> sink,
    //   其 next 方法在每次回调时只能被调用一次.
    @Test
    public void testStateBasedGenerate() {
        Flux<String> flux = Flux.generate(
                // You have to provide a Supplier<S> for the initial state, and your generator function now returns a new state on each round.
                // We supply the initial state value of 0
                () -> 0,
                (state, sink) -> {
                    // We use the state to choose what to emit (a row in the multiplication table of 3)
                    sink.next("3 x " + state + " = " + 3 * state);
                    // We also use it to choose when to stop
                    if (state == 10) sink.complete();
                    // We return a new state that we use in the next invocation (unless the sequence terminated in this one).
                    return state + 1;
                });
        flux.subscribe(System.out::println);
    }

    @Test
    public void testMutableStateVariantOfGenerateMethod() {
        Flux<String> flux = Flux.generate(
                AtomicLong::new,
                (state, sink) -> {
                    long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3 * i);
                    if (i == 10) sink.complete();
                    return state;
                });
        flux.subscribe(System.out::println);
    }

    @Test
    public void testGenerateWithConsumer() {
        Flux<String> flux = Flux.generate(
                AtomicLong::new,
                (state, sink) -> {
                    long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3 * i);
                    if (i == 10) sink.complete();
                    return state;
                },
                // Do some cleanups here.
                (state) -> System.out.println("state: " + state)
        );
        flux.subscribe();
    }

    // =================================================================================================================

    // ~ 4.4.2 Asynchronous and Multi-threaded: create
    //   https://projectreactor.io/docs/core/release/reference/#producing.create

    //   create 方法是编程式创建 Flux 方法中的比较进阶的一种方式. 它支持一个在循环中生产多个数据, 甚至是从多线程中.
    //      它暴露了一个 FluxSink, 提供 next, error 和 complete 方法.
    //      与 generate 相比, 它没有基于状态 (state-based) 的重载. 另外, 它会在回调中触发多线程事件.
    //   create 方法的适用场景之一是将已有的 API 与响应式 API 连接起来. 比如一个基于 Listener 的异步 API.

    //   create 方法并不会将代码并行化也不会让代码异步执行, 尽管如此, 它却能够被用于异步 APIs.
    //   如果你的代码在 create 方法的 lambda 表达式参数中阻塞了, 程序将会死锁.
    //   Even with the use of subscribeOn,
    //   there’s the caveat that a long-blocking create lambda (such as an infinite loop calling sink.next(t)) can lock the pipeline:
    //   the requests would never be performed due to the loop starving the same thread they are supposed to run from.
    //   Use the subscribeOn(Scheduler, false) variant:
    //   requestOnSeparateThread = false will use the Scheduler thread for the create and still let data flow by performing request in the original thread.

    // =================================================================================================================

    // ~ 4.4.3 Asynchronous but single-threaded: push
    //   https://projectreactor.io/docs/core/release/reference/#_asynchronous_but_single_threaded_push
    //   push 介于 generate 和 create 之间, 适合于从单一生产者产生事件. "同一时间只有一个生产者" 线程可以调用 next, complete 或 error

    //   ~ A hybrid push/pull model
    //     https://projectreactor.io/docs/core/release/reference/#_a_hybrid_pushpull_model

    //   ~ Cleaning up after push() or create()
    //     https://projectreactor.io/docs/core/release/reference/#_cleaning_up_after_push_or_create

    // ~ 4.4.4 Handle
    //   https://projectreactor.io/docs/core/release/reference/#_cleaning_up_after_push_or_create

    /**
     * Description: handle 是一个实例方法, 意味着它用于连接已经存在的序列. 在 {@link Mono} 和 {@link Flux} 中都有提供.<br>
     * Details: 和 generate 类似, 它使用 {@link reactor.core.publisher.SynchronousSink} 并且只允许一个接一个的生产. 然而, Handle 能够用于实现类似 filter 和 map 的效果.
     *
     * @return void
     * @author LiKe
     * @date 2020-08-20 20:08:17
     */
    @Test
    public void testHandle() {
        Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20)
                .handle((i, sink) -> {
                    String letter = alphabet(i);
                    // Filter it out by not calling sink.next.
                    if (letter != null)
                        sink.next(letter);
                });

        alphabet.subscribe(System.out::println);
    }

    private String alphabet(int letterNumber) {
        if (letterNumber < 1 || letterNumber > 26) {
            return null;
        }
        int letterIndexAscii = 'A' + letterNumber - 1;
        return "" + (char) letterIndexAscii;
    }


}
